package com.framework.websocket.core;

import com.framework.websocket.model.SubscribeMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.websocket.Session;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * 通用消息处理器
 * 处理所有类型的订阅和消息推送
 */
@Slf4j
@Component
public class GenericMessageHandler {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private SubscriptionConfig subscriptionConfig;
    
    @Autowired
    private WebSocketManager webSocketManager;
    
    /**
     * 订阅管理：订阅类型 -> 会话列表
     */
    private final Map<String, List<Session>> subscriptions = new ConcurrentHashMap<>();
    
    /**
     * 处理订阅请求
     */
    public void subscribe(Session session, String subscriptionType) {
        if (!subscriptionConfig.isValidSubscriptionType(subscriptionType)) {
            log.warn("无效的订阅类型: {}", subscriptionType);
            return;
        }
        
        List<Session> sessions = subscriptions.computeIfAbsent(subscriptionType, 
            k -> new CopyOnWriteArrayList<>());
        
        if (!sessions.contains(session)) {
            sessions.add(session);
            log.info("会话 {} 订阅了类型: {}", session.getId(), subscriptionType);
            
            // 发送订阅成功消息
            Map<String, Object> response = new HashMap<>();
            response.put("message", "订阅成功");
            response.put("type", subscriptionType);
            webSocketManager.sendMessage(session, "SUBSCRIBED", subscriptionType, response);
            
            // 推送初始数据
            pushInitialData(session, subscriptionType);
        }
    }
    
    /**
     * 取消订阅
     */
    public void unsubscribe(Session session, String subscriptionType) {
        List<Session> sessions = subscriptions.get(subscriptionType);
        if (sessions != null) {
            sessions.remove(session);
            log.info("会话 {} 取消订阅类型: {}", session.getId(), subscriptionType);
        }
    }
    
    /**
     * 清除会话的所有订阅
     */
    public void clearSubscriptions(Session session) {
        subscriptions.values().forEach(sessions -> sessions.remove(session));
        log.info("清除会话 {} 的所有订阅", session.getId());
    }
    
    /**
     * 处理 Redis 键变化事件
     */
    public void handleKeyChange(String key) {
        // 根据键找到对应的订阅类型
        String subscriptionType = subscriptionConfig.getSubscriptionType(key);
        if (subscriptionType == null) {
            log.debug("键 {} 没有对应的订阅类型", key);
            return;
        }
        
        // 获取数据
        Object data = redisTemplate.opsForValue().get(key);
        if (data == null) {
            log.debug("键 {} 的数据为空", key);
            return;
        }
        
        // 推送给所有订阅者
        List<Session> sessions = subscriptions.get(subscriptionType);
        if (sessions != null && !sessions.isEmpty()) {
            for (Session session : sessions) {
                if (session.isOpen()) {
                    // 使用新的推送方法，明确传递主题
                    webSocketManager.sendPushMessage(session, subscriptionType, key, data);
                }
            }
            log.debug("推送 {} 类型消息给 {} 个订阅者", subscriptionType, sessions.size());
        }
    }
    
    /**
     * 推送初始数据
     */
    private void pushInitialData(Session session, String subscriptionType) {
        String keyPrefix = subscriptionConfig.getKeyPrefix(subscriptionType);
        if (keyPrefix == null) {
            return;
        }
        
        // 根据不同的订阅类型，推送不同的初始数据
        switch (subscriptionType) {
            case SubscriptionConfig.TYPE_DEVICE_STATUS:
                // 推送所有设备的当前状态
                pushAllKeysWithPrefix(session, subscriptionType, keyPrefix);
                break;
                
            case SubscriptionConfig.TYPE_ALARM:
                // 推送最近的告警信息
                pushRecentAlarms(session, subscriptionType, keyPrefix);
                break;
                
            default:
                // 其他类型暂不推送初始数据
                log.debug("订阅类型 {} 不需要推送初始数据", subscriptionType);
        }
    }
    
    /**
     * 推送指定前缀的所有键值
     */
    private void pushAllKeysWithPrefix(Session session, String subscriptionType, String keyPrefix) {
        try {
            // 使用 scan 命令查找所有匹配的键
            redisTemplate.keys(keyPrefix + "*").forEach(key -> {
                Object value = redisTemplate.opsForValue().get(key);
                if (value != null) {
                    // 使用新的推送方法
                    webSocketManager.sendPushMessage(session, subscriptionType, key, value);
                }
            });
        } catch (Exception e) {
            log.error("推送初始数据失败", e);
        }
    }
    
    /**
     * 推送最近的告警信息
     */
    private void pushRecentAlarms(Session session, String subscriptionType, String keyPrefix) {
        // 这里可以根据业务需求实现
        // 例如：从 Redis 的有序集合中获取最近的告警
        log.debug("推送最近的告警信息给会话 {}", session.getId());
    }
    
    /**
     * 获取订阅统计信息
     */
    public Map<String, Integer> getSubscriptionStats() {
        Map<String, Integer> stats = new HashMap<>();
        subscriptions.forEach((type, sessions) -> {
            stats.put(type, sessions.size());
        });
        return stats;
    }
}